package blue.mqtt4j.core.handler;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * @author Jin Zheng
 * @since 2019-05-02
 */
public class HeartbeatHandler extends SimpleChannelInboundHandler<MqttMessage>
{
	private static Logger logger = LoggerFactory.getLogger(HeartbeatHandler.class);

	private InetSocketAddress remoteAddress;

	public HeartbeatHandler()
	{
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttMessage message) throws Exception
	{
		MqttFixedHeader reqHeader = message.fixedHeader();
		MqttMessageType type = reqHeader.messageType();
		if (type == MqttMessageType.PINGREQ)
		{
			if (logger.isDebugEnabled())
			{
				logger.debug("Receive PING from {}: {}", remoteAddress, JSON.toJSONString(reqHeader));
			}
			ctx.writeAndFlush(MqttMessage.PINGRESP);
		}
		else if (type == MqttMessageType.DISCONNECT)
		{
			logger.info("disconnect from: {}", ctx.channel().remoteAddress());
			ctx.channel().close();
		}
		else
		{
			ReferenceCountUtil.retain(message);
			ctx.fireChannelRead(message);
		}
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception
	{
		if (evt instanceof IdleStateEvent)
		{
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state() == IdleState.READER_IDLE)
			{
				if (logger.isDebugEnabled())
				{
					logger.debug("Read idle, no receive PONG, close!");
				}
				ctx.close();
			}
			else if (event.state() == IdleState.WRITER_IDLE)
			{
				ctx.writeAndFlush(MqttMessage.PINGREQ);
				if (logger.isDebugEnabled())
				{
					logger.debug("Send PING  to {}", remoteAddress);
				}
			}
		}

		super.userEventTriggered(ctx, evt);
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{
		Channel ch = ctx.channel();
		remoteAddress = (InetSocketAddress) ch.remoteAddress();

		super.channelActive(ctx);
	}
}
